package com.teamsun.mqtt.service;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.stream.Stream;

import org.apache.log4j.Logger;
import org.springframework.beans.factory.annotation.Autowired;

import com.teamsun.mqtt.entity.MsgRep;
import com.teamsun.mqtt.entity.SendMsg;
import com.teamsun.mqtt.entity.vo.DelayWillRemoveMsg;
import com.teamsun.mqtt.entity.vo.SendableMsg;
import com.teamsun.mqtt.handle.PushServiceHandle;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufAllocator;
import io.netty.handler.codec.mqtt.MqttQoS;

/**
 * 获取未读消息
 * 
 * @author acer
 *
 */
public class GetUnReadData {

	Logger logger = Logger.getLogger(getClass());

	BlockingQueue<MsgRep> msgRepsque;

	@Autowired
	MsgService msgService;
	
	
	@Autowired
	MessagePushService messagePushService;

	@Autowired
	PushServiceHandle pubservice;


	public GetUnReadData() {

	}

	/**
	 * 发送未读
	 */
	public void sendUnReadMsg(String clientid,ByteBufAllocator allocator) throws Exception {

		List<MsgRep> msgReps = queryMsgTabGetMsg(clientid);
		
		if(msgReps==null)return;
		
		Stream<MsgRep> stream=msgReps.parallelStream();
		
		stream.forEach((MsgRep msgRep)->{
			
			ByteBuf sendBytebuf = null;
			if (msgRep.getContent() != null) {
				sendBytebuf = createNewBuff(msgRep.getContent(),allocator);
			}
			
			if (sendBytebuf != null && sendBytebuf.refCnt() > 0) {
				SendableMsg sendableMsg = new SendableMsg();
				sendableMsg.setDupTimes(0);
				sendableMsg.setMessageid(msgRep.getId());
				sendableMsg.setMsgContent(sendBytebuf);
				sendableMsg.setQoS(MqttQoS.EXACTLY_ONCE);
				sendableMsg.setSendclientid(msgRep.getSendiden());
				sendableMsg.setTopname(msgRep.getTopname());
				sendableMsg.setRecvclientid(clientid);
				messagePushService.sendMsg2Client(sendableMsg);

			}
		});
		

	}

	/**
	 * 根据一个字节数组创建一个bytebuf
	 * 
	 * @param bs
	 * @return
	 */
	public ByteBuf createNewBuff(byte[] bs,ByteBufAllocator allocator) {
		ByteBuf sendBytebuf = allocator.buffer();
		sendBytebuf.writeBytes(bs);
		sendBytebuf.resetReaderIndex();
		return sendBytebuf;
	}

	/**
	 * 获取未读消息 未读消息状态都是存储在内存里面 {@link msgSendedQue} 从{@link msgSendedQue}
	 * 里面拿到未读状态消息的id。 分别从数据库和消息持久缓存{@link msgRepsque} 根据这些id获取消息。
	 * 
	 * @param session
	 * @return
	 */
	public List<MsgRep> queryMsgTabGetMsg(String clientid) {


		List<Integer> msgids = new ArrayList<Integer>();
		
		Stream<SendMsg> streams=
				messagePushService.getUnsucQue().parallelStream();
		List<SendMsg> sendMsgRemoves=new ArrayList<>();
		streams.forEach((SendMsg sendMsg)->{
			if (sendMsg.getClientid().equals(clientid)) {
				sendMsgRemoves.add(sendMsg);
				msgids.add(sendMsg.getSendmsgid());
			}
		});
		streams.close();
		messagePushService.getUnsucQue().removeAll(sendMsgRemoves);
		
		if (msgids.isEmpty())
			return null;

		Map<String, Object> map = new HashMap<String, Object>();
		map.put("list", msgids);
		final List<MsgRep> result =new ArrayList<>();
		
		try {
			result.addAll(msgService.findMsgs(msgids));
		} catch (Exception e) {
			 e.printStackTrace();
		}
		
	
		Stream<MsgRep> streamRep=msgRepsque.parallelStream();
		streamRep.forEach((MsgRep msgRep)->{
			if (msgids.contains(msgRep.getId())) {
				result.add(msgRep);
			}
		});
		
		streamRep.close();

		return result;
	}

	public BlockingQueue<MsgRep> getMsgRepsque() {
		return msgRepsque;
	}

	public void setMsgRepsque(BlockingQueue<MsgRep> msgRepsque) {
		this.msgRepsque = msgRepsque;
	}

	

}
